package com.open.source;

import static com.open.source.MqConstant.PRODUCER_APPLICATION_NAME;
import static com.open.source.MqConstant.PRODUCER_HOST_IP;
import static com.open.source.MqSwitchControlConditionInitializer.DEFAULT_MQ_SELECTED;
import static com.open.source.MqSwitchControlConditionInitializer.KAFKA;
import static com.open.source.MqSwitchControlConditionInitializer.RABBIT_MQ;

import com.open.source.MqManagerEnum.RabbitRouting;
import com.open.source.kafka.ApplicationContextHolder;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import lombok.SneakyThrows;
import org.apache.commons.collections4.MapUtils;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AlterConfigOp;
import org.apache.kafka.clients.admin.AlterConfigOp.OpType;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.config.ConfigResource.Type;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.kafka.core.KafkaTemplate;

/**
 * @author ZonLen since on 2022/8/21 上午1:20
 */
public final class MqSenderAdaptor {

  private static final List<String> CREATED_TOPIC = new ArrayList<>();

  /**
   * 发送消息
   */
  public static void send(MqManagerEnum kafkaTopic, byte[] data) {
    send(kafkaTopic, data, StandardCharsets.UTF_8.name());
  }

  @SneakyThrows
  public static void send(MqManagerEnum kafkaTopic, byte[] data, String charset) {
    String selectedMq = ApplicationContextHolder.getProperty(DEFAULT_MQ_SELECTED, RABBIT_MQ);
    String appName = ApplicationContextHolder.getProperty("spring.application.name");
    if (KAFKA.equalsIgnoreCase(selectedMq)) {
      final String topic = ApplicationContextHolder.getEnvironment()
          .resolvePlaceholders(kafkaTopic.getTopic());
      final ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic,
          null, System.currentTimeMillis(), null, data,
          headers(appName));
      ApplicationContextHolder.getBean(KafkaTemplate.class).send(record);
    } else {
      final RabbitRouting rabbitRouting = kafkaTopic.getRabbitRouting();
      //todo header 改造
      final MessageProperties messageProperties = new MessageProperties();
//      messageProperties.setHeader("corpid", InvocationInfoUtils.getCorpId());
//      messagePropertie
//          .setHeader(Constants.REQUEST_ID_KEY, MDC.get(Constants.REQUEST_ID_KEY));
      Message message = new Message(data, messageProperties);
      ApplicationContextHolder.getBean(RabbitTemplate.class)
          .send(rabbitRouting.getExchange(), rabbitRouting.getRoutingKey(), message);
    }
  }

  //todo header 改造
  private static Iterable<Header> headers(String appName) throws UnknownHostException {
    final List<Header> headers = new ArrayList<>();
    headers
        .add(new RecordHeader(PRODUCER_HOST_IP,
            InetAddress.getLocalHost().getHostAddress().getBytes()));
//    headers.add(new RecordHeader(Constants.REQUEST_ID_KEY, MDC.get(Constants.REQUEST_ID_KEY)));
    headers.add(new RecordHeader(PRODUCER_APPLICATION_NAME, appName.getBytes()));
    headers.add(new RecordHeader("trance-id", UUID.randomUUID().toString().getBytes()));
    headers.add(new RecordHeader(MqConstant.X_RETRIES_HEADER, "0".getBytes()));
//    headers.add(new RecordHeader("corpid", InvocationInfoUtils.getCorpId()));
    return headers;
  }


  public static void alertTopicConfig(String topic, Map<String, String> config) {
    final List<AlterConfigOp> alterConfigOps = new ArrayList<>();
    MapUtils.emptyIfNull(config).forEach(
        (k, v) -> alterConfigOps.add(new AlterConfigOp(new ConfigEntry(k, v), OpType.SET)));
    final HashMap<ConfigResource, Collection<AlterConfigOp>> configResource = new HashMap<>();
    configResource.put(new ConfigResource(Type.TOPIC, topic), alterConfigOps);
    final Map<String, Object> adminProperties = ApplicationContextHolder
        .getBean(KafkaProperties.class)
        .buildAdminProperties();
    Admin.create(adminProperties)
        .incrementalAlterConfigs(configResource);
  }

  /**
   * 创建topic
   */
  public static void createTopicConfig(String topic) {
    final Map<String, String> config = new HashMap<>();
    config.put(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(720 * 3600));
    createTopicConfig(topic, config);
  }

  /**
   * 创建topic
   */
  public static void createTopicConfig(String topic, Map<String, String> config) {
    if (CREATED_TOPIC.contains(topic)) {
      return;
    }
    final Map<String, Object> adminProperties = ApplicationContextHolder
        .getBean(KafkaProperties.class)
        .buildAdminProperties();
    final NewTopic newTopic = new NewTopic(topic, Optional.empty(), Optional.empty())
        .configs(config);
    final boolean done = Admin.create(adminProperties).createTopics(Collections.singleton(newTopic))
        .config(topic).isDone();
    if (done) {
      CREATED_TOPIC.add(topic);
    }
  }

}
